package com.test;

import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.logging.LogLevel;
import org.pentaho.di.core.util.EnvUtil;
import org.pentaho.di.job.Job;
import org.pentaho.di.job.JobMeta;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;

import java.util.HashMap;
import java.util.Map;

/**
 * Created with IDEA
 * User: LEIJIE
 * Date: 2017/5/24 0024
 * Time: 15:15
 */
public class test4 {

    public static void main(String[] args) {
        Map<String,String> tmp = new HashMap<>();
        tmp.put("name","雷2杰");
        tmp.put("pwd","123");
        tmp.put("jsonstr","{\"name\":\"leijie\",\"pwd\":\"123\"}");
//        runTrans(null,"G:\\etltest\\jsonTra.ktr");
        runJob(tmp,"G:\\fileRepository\\jsonJob.kjb");
    }

    /**
     * 调用 kettle trans
     *
     * @param params 多个参数值
     * @param ktrPath  如： String fName= "D:\\kettle\\aaa.ktr";
     */
    public static void runTrans(String[] params, String ktrPath) {
        try {
            KettleEnvironment.init();
            EnvUtil.environmentInit();
            TransMeta transMeta = new TransMeta(ktrPath);
            Trans trans = new Trans(transMeta);
//            trans.setVariable("jsonstr","{\"name\":\"leijie\",\"pwd\":\"123\"}");
//            trans.setParameterValue("jsonstr","{\"name\":\"leijieddd\",\"pwd\":\"12d3\"}");
            trans.execute(params);
            trans.waitUntilFinished();
            if (trans.getErrors() > 0) {
                throw new Exception("Errors during transformation execution!");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 调用 kettle job
     *
     * @param jobPath 如： String fName= "D:\\kettle\\aaa.kjb";
     */
    public static void runJob(Map<String,String> tmp, String jobPath) {
        try {
            KettleEnvironment.init();
            JobMeta jobMeta = new JobMeta(jobPath, null);
            Job job = new Job(null, jobMeta);
            // 向Job 脚本传递参数，脚本中获取参数值：${参数名}
            if (tmp != null && !tmp.isEmpty()) {
                tmp.forEach((k,v)->{
                    job.setVariable(k, v);
                });
            }
            job.setLogLevel(LogLevel.DEBUG);
            job.start();
            job.waitUntilFinished();
            if (job.getErrors() > 0) {
                throw new Exception("Errors during job execution!");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
